package com.example.tableSql

import java.util.Date

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment


object TimeWindowDemo2 {

  case class Supp(name: String, part: String, number: String, cost:String,timestamp:Long)

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val tEnv = StreamTableEnvironment.create(env)

    val dsStr: DataStream[String] = env.readTextFile("E:\\note\\flink-project\\temp\\supplie_timestamp.txt")

    val sDs: DataStream[Supp] = dsStr.map(data => {
      val arrs: Array[String] = data.split(",")
      Supp(arrs(0), arrs(1), arrs(2).toString, arrs(3), arrs(4).toLong)
    })
//      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Supp](Time.seconds(1)) {
//      override def extractTimestamp(element: Supp): Double = element.timestamp * 1000L
//    })

    //    示例二
    val table: Table = tEnv.fromDataStream(sDs)
    table.printSchema()

    env.execute()
  }
}
